package com.devilvan.customSort;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class CustomSortReducer extends Reducer<FlowSortBean,NullWritable,Text, Text> {
    private Text text;

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        text = new Text();
    }

    public enum ReduceEnum {
        CUSTOM_COUNTER_REDUCE("COUNTER_REDUCER_GROUP","COUNTER_REDUCER_NAME");

        private final String group;
        private final String name;

        ReduceEnum(String group, String name) {
            this.group = group;
            this.name = name;
        }

        public String getGroup() {
            return group;
        }

        public String getName() {
            return name;
        }
    }

    @Override
    protected void reduce(FlowSortBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
        context.getCounter(ReduceEnum.CUSTOM_COUNTER_REDUCE.getGroup(),ReduceEnum.CUSTOM_COUNTER_REDUCE.getName()).increment(1L);

        text.set(key.getUpPackNum() + "\t" +
                key.getDownPackNum() + "\t" +
                key.getUpPayLoad() + "\t" +
                key.getDownPayLoad());
        context.write(new Text(key.getMsiSdn()),text);
    }
}
